package com.ds.infrastructure.hyperspace.container.endpoint.connector.health;

import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.Charsets;
import org.springframework.util.StringUtils;

import static com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum.HTTPS;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.EMPTY_STR;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http1HealthCheckHandler extends SimpleChannelInboundHandler {

    private final TargetEndpointNode targetEndpointNode;

    private final HyperspaceNodeHealthChecker nodeHealthChecker;

    private final HttpHealthCheckChannelFutureListener httpHealthCheckFutureListener;

    public Http1HealthCheckHandler(TargetEndpointNode targetEndpointNode,
            HyperspaceNodeHealthChecker nodeHealthChecker) {
        this.targetEndpointNode = targetEndpointNode;
        this.nodeHealthChecker = nodeHealthChecker;
        this.httpHealthCheckFutureListener = new HttpHealthCheckChannelFutureListener();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        TargetNodeConfig nodeConfig = targetEndpointNode.getNodeConfig();
        if (nodeConfig.getProtocol().getIndex() > HTTPS.getIndex()) {
            targetEndpointNode.incrementPassCount(ctx.channel().remoteAddress());
            ctx.flush().close();
            return;
        }

        String healthCheckUri = nodeConfig.getProtocolConfig().getHealthCheckConfig().getUri();
        if (!StringUtils.isEmpty(healthCheckUri)) {
            HttpRequest request = new DefaultFullHttpRequest(
                    HttpVersion.HTTP_1_1, HttpMethod.GET, healthCheckUri);
            request.headers().set(HttpHeaderNames.HOST, targetEndpointNode.getHostHeader());
            request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
            ctx.writeAndFlush(request).addListener(httpHealthCheckFutureListener);
        } else {
            targetEndpointNode.incrementPassCount(ctx.channel().remoteAddress());
            ctx.flush().close();
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            if (!(msg instanceof FullHttpResponse)) {
                return;
            }
            FullHttpResponse healthCheckResponse = (FullHttpResponse)msg;
            String expect = targetEndpointNode.getNodeConfig().getProtocolConfig()
                    .getHealthCheckConfig().getExpect();
            boolean isPass = true;
            if (healthCheckResponse.status().code() >= HttpResponseStatus.BAD_REQUEST.code()){
                isPass = false;
            }
            if (isPass && !StringUtils.isEmpty(expect)) {
                String responseStr = EMPTY_STR;
                byte[] responseBody = ByteBufUtil.getBytes(healthCheckResponse.content());
                if (responseBody != null) {
                    responseStr = new String(responseBody, Charsets.UTF_8);
                }
                if (!expect.equals(responseStr)) {
                    isPass = false;
                }
            }
            if (isPass) {
                targetEndpointNode.incrementPassCount(ctx.channel().remoteAddress());
            } else {
                targetEndpointNode.incrementFailCount(ctx.channel().remoteAddress());
            }
        } finally {
            ctx.flush().close();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
        nodeHealthChecker.checkComplete();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Http1HealthCheckHandler failed!", cause);
        Channel channel = ctx.channel();
        if (channel.isOpen()) {
            channel.flush().close();
        }
    }

    class HttpHealthCheckChannelFutureListener implements ChannelFutureListener {

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                log.error("Http health check failed, target node ({}),remote address:{}",
                        targetEndpointNode.getKey(),future.channel().remoteAddress(), future.cause());
                targetEndpointNode.incrementFailCount(future.channel().remoteAddress());
                future.channel().close();
            }
        }
    }

}
